package com.example.startup;

import com.example.config.SysMQConfig;
import com.example.constant.MQConstant;
import com.example.dto.MQListenerDto;
import com.example.listener.SysMQMessageListener;
import com.example.listener.handler.SysMQMessageListenerHandler;
import com.example.message.MqMessageReturn;
import com.example.service.SysServerService;
import lombok.Data;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.connection.CachingConnectionFactory;

import javax.jms.*;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Data
public class SysActiveMQStartUpService extends AbstractMQStartUpService {

    Logger logger = LoggerFactory.getLogger(getClass());

    private volatile boolean running = true;

    private SysMQConfig sysMQConfig;

    private SysMQMessageListenerHandler sysMQMessageListenerHandler;

    private RedeliveryPolicy redeliveryPolicy;

    private ActiveMQConnectionFactory activeMQConnectionFactory;

    private CachingConnectionFactory connectionFactory;

    private Connection connection;

    private ExecutorService executorService;

    private SysServerService sysServerService;

    /****
     * 扫描注解并启动监听线程
     * @param sysMQConfig
     */
    @Override
    public void init(SysMQConfig sysMQConfig, SysMQMessageListenerHandler sysMQMessageListenerHandler , SysServerService sysServerService) {
        try {
            this.executorService = Executors.newCachedThreadPool();
            this.sysMQConfig = sysMQConfig;
            this.sysMQMessageListenerHandler = sysMQMessageListenerHandler;
            this.sysServerService = sysServerService;
            this.redeliveryPolicy = new RedeliveryPolicy();
            this.redeliveryPolicy.setMaximumRedeliveries(this.sysMQConfig.getRedeliverySize());
            this.redeliveryPolicy.setRedeliveryDelay(this.sysMQConfig.getRedeliveryDelay());
            this.activeMQConnectionFactory = new ActiveMQConnectionFactory(sysMQConfig.getUsername(), sysMQConfig.getPassword(), "tcp://" + sysMQConfig.getHost() + ":" + sysMQConfig.getPort());
            this.activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);
            this.connectionFactory = new CachingConnectionFactory(activeMQConnectionFactory);
            this.connection = this.connectionFactory.createConnection();
            this.connection.start();
        } catch (JMSException e) {
            logger.warn("ActiveMQ服务 启动失败：｛｝", e);
            if (this.connection != null) {
                try {
                    this.connection.close();
                } catch (JMSException ex) {
                    logger.warn("关闭connection异常: " + ex.getMessage());
                }
            }
        }
    }

    /****
     * 注册消费线程并拉取数据
     * @throws Exception
     */
    @Override
    public void start() throws Exception {
        Map<String, MQListenerDto> listenerMap = this.sysMQMessageListenerHandler.getListenerMap();
        for (String name : listenerMap.keySet()) {
            MQListenerDto mqListenerDto = listenerMap.get(name);
            executorService.execute(new MQListenerThread(mqListenerDto));
        }
    }

    /****
     * 关闭连接回收资源
     * @throws Exception
     */
    @Override
    public void stop() throws Exception {
        if (this.connection != null) {
            try {
                this.connection.close();
            } catch (JMSException e) {
                logger.warn("close connection error", e);
            }
        }
        this.executorService.shutdownNow();
    }

    /****
     * 发送消息并关闭资源
     * @param type
     * @param name
     * @param message
     * @throws Exception
     */
    @Override
    public void send(String type, String name, String message) throws Exception {
        Session session = null;
        MessageProducer messageProducer = null;
        try {
            session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            if (type.equalsIgnoreCase(MQConstant.QUEUE)) {
                TextMessage textMessage = session.createTextMessage(message);
                Queue queue = session.createQueue(name);
                messageProducer = session.createProducer(queue);
                messageProducer.send(textMessage);
            }
            if (type.equalsIgnoreCase(MQConstant.TOPIC)) {
                TextMessage textMessage = session.createTextMessage(message);
                Topic topic = session.createTopic(name);
                messageProducer = session.createProducer(topic);
                messageProducer.send(textMessage);
            }
        } catch (Exception e) {
            logger.warn("MQ推送消息失败: " + e.getMessage());
            throw e;
        } finally {
            if (messageProducer != null) {
                messageProducer.close();
            }
            if (session != null) {
                session.close();
            }
        }
    }

    @Data
    class MQListenerThread implements Runnable {

        private MQListenerDto mqListenerDto;

        public MQListenerThread(MQListenerDto mqListenerDto) {
            this.mqListenerDto = mqListenerDto;
        }

        @Override
        public void run() {
            MessageConsumer messageConsumer = null;
            Session session = null;
            try {
                if (mqListenerDto.getType().equalsIgnoreCase(MQConstant.QUEUE) && mqListenerDto.getMqListener().repeated()) {
                    session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
                } else {
                    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                }
                if (mqListenerDto.getType().equalsIgnoreCase(MQConstant.QUEUE)) {
                    Queue queue = session.createQueue(mqListenerDto.getName());
                    messageConsumer = session.createConsumer(queue);
                }
                if (mqListenerDto.getType().equalsIgnoreCase(MQConstant.TOPIC)) {
                    Topic topic = session.createTopic(mqListenerDto.getName());
                    messageConsumer = session.createConsumer(topic);
                }
                while (running) {
                    try {
                        Message message = messageConsumer.receive();
                        if (message != null && message instanceof TextMessage) {
                            SysMQMessageListener sysMQMessageListener = (SysMQMessageListener) sysMQMessageListenerHandler.getApplicationContext().getBean(mqListenerDto.getBeanName());
                            executorService.execute(new MQMessageReceiveThread(mqListenerDto, sysMQMessageListener, (TextMessage) message, session));
                        }
                    } catch (Exception e) {
                        if (!(e instanceof InterruptedException || e instanceof JMSException)) {
                            logger.warn("监听MQ消息线程失败: " + e.getMessage());
                        }
                    }
                }
            } catch (Exception e) {
                if (!(e instanceof InterruptedException || e instanceof JMSException)) {
                    logger.warn("监听MQ消息线程失败: " + e.getMessage());
                }
            } finally {
                if (messageConsumer != null) {
                    try {
                        messageConsumer.close();
                    } catch (JMSException e) {
                        logger.warn("关闭消息接受出错: " + e.getMessage());
                    }
                }
                if (session != null) {
                    try {
                        session.close();
                    } catch (JMSException e) {
                        logger.warn("关闭session异常: " + e.getMessage());
                    }
                }
            }
        }
    }

    //真正的消息处理线程
    class MQMessageReceiveThread implements Runnable {

        private MQListenerDto mqListenerDto;

        private SysMQMessageListener sysMQMessageListener;

        private TextMessage textMessage;

        private Session session;

        public MQMessageReceiveThread(MQListenerDto mqListenerDto, SysMQMessageListener sysMQMessageListener, TextMessage textMessage, Session session) {
            this.mqListenerDto = mqListenerDto;
            this.sysMQMessageListener = sysMQMessageListener;
            this.textMessage = textMessage;
            this.session = session;
        }

        @Override
        public void run() {
            MqMessageReturn messageReturn = null;
            try {
                messageReturn = sysMQMessageListener.onMessage(textMessage.getText());
            } catch (Exception e) {
                messageReturn = MqMessageReturn.error(e.getMessage());
            } finally {
                if (mqListenerDto.getType().equalsIgnoreCase(MQConstant.QUEUE) && mqListenerDto.getMqListener().repeated()) {
                    if (messageReturn.isSuccess()) {
                        try {
                            textMessage.acknowledge();
                        } catch (JMSException e) {
                            logger.warn("textMessage acknowledge error : " + e.getMessage());
                        }
                    } else {
                        try {
                            session.recover();
                        } catch (JMSException e) {
                            logger.warn("session recover error : " + e.getMessage());
                        }
                    }
                }
            }
        }
    }
}
